package com.huawei.dli.flink.sye;

import com.huawei.dli.flink.sye.sink.HBaseSinkAvro;
import com.huawei.dli.flink.sye.source.KafkaKerbosSource;
import com.huawei.dli.flink.sye.utils.KafkaUtil;
import com.huawei.hwclouds.drs.avro.Record;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class App2HbaseAvro {
    public static void main(String[] args) throws Exception {
        // 获取参数
        ParameterTool params = ParameterTool.fromArgs(args);

        final StreamExecutionEnvironment env = StreamExecutionEnvironment
                .getExecutionEnvironment();

        env.enableCheckpointing(5000);
        env.setParallelism(5);

        String userPrincipal = params.getRequired("principal");
        KafkaKerbosSource<Record> kafkaAvroSource = KafkaUtil.getKafkaAvroSource(params,userPrincipal);

        env.addSource(kafkaAvroSource,"kafka_source").disableChaining()
                .flatMap(new FlatMapFunction<Record, String>() {
                    @Override
                    public void flatMap(Record value, Collector<String> out) throws Exception {
                        out.collect(value.toString());
                    }
                }).disableChaining()
                .addSink(new HBaseSinkAvro(userPrincipal));

        env.execute("bigdatadws");
    }
}
